/*
package com.audaque.springboot.foshanupload.reactor.webfluxdemo.controller;


import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

*/
/*
 *
 * @author zgb
 * @desc 依赖jdk11 ,消息发布订阅
 * @date 2022-11-24 16:20:35

 *//*


@Slf4j
@RestController
@RequestMapping("flow")
public class FlowController {
    */
/**
     * Subscriber 会将 Publisher 发布的数据缓存在 Subscription 中，其长度默认为 256，一旦超出这个数据量，publisher 就会降低数据发送速度。
     *//*

    @RequestMapping("backPressure")
    public void backPressure() {

        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();

        Flow.Subscriber<String> subscriber = new Flow.Subscriber<String>() {
            private Flow.Subscription subscription;

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                this.subscription = subscription;
                //向数据发布者请求一个数据
                this.subscription.request(1);
            }

            @Override
            public void onNext(String item) {
                System.out.println("接收到 publisher 发来的消息了：" + item);
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                this.subscription.request(1);
            }

            @Override
            public void onError(Throwable throwable) {
                //出现异常，就会来到这个方法，此时直接取消订阅即可
                this.subscription.cancel();
            }

            @Override
            public void onComplete() {
                //发布者的所有数据都被接收，并且发布者已经关闭
                System.out.println("数据接收完毕");
            }
        };
        publisher.subscribe(subscriber);
        for (int i = 0; i < 500; i++) {
            System.out.println("i--------->" + i);
            publisher.submit("hello:" + i);
        }
        //关闭发布者
        publisher.close();
    }


    @RequestMapping("pubSub")
    public void pubsub() {
        //生产者
        SubmissionPublisher publisher = new SubmissionPublisher<>();
        //订阅者
        Flow.Subscriber subscriber = new Flow.Subscriber() {
            */
/**
             * 定义成员变量来共享订阅关系，包含 request(long n);和 cancel()
             * request(long n)：每次消息到来，订阅者会读取订阅关系，确认能接收多少消息，接收后自动减一
             * cancel()：取消订阅;
             *
             *//*


            Flow.Subscription subscription;

            */
/**
             * 建立关系
             * @param subscription a new subscription
             *//*

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                //线程名称：ForkJoinPool.commonPool-worker-19 ，可以看出底层使用了ForkJoinPool线程池
                log.debug(Thread.currentThread().getName() + "：发布者与订阅者建立订阅关系，只会执行一次");
                //把订阅关系赋值给成员变量
                this.subscription = subscription;
                //建立关系时设置接收数量大于1代表开始接收消息，设置小于1会报错
                this.subscription.request(1);
            }

            @Override
            public void onNext(Object item) {
                //继续接收则设置接收数量大于1
                this.subscription.request(1);
                //不再接收消息则调用cancel
                //this.subscription.cancel();
                log.debug(Thread.currentThread().getName() + "：receive :  " + item);
            }

            @Override
            public void onError(Throwable throwable) {
                //出现异常
                throwable.printStackTrace();
                //不再接收消息则调用cancel
                this.subscription.cancel();
                log.debug(Thread.currentThread().getName() + "： onError ");
            }

            @Override
            public void onComplete() {
                //全部处理完（发布者关闭）
                log.debug("处理完成");
                log.debug(Thread.currentThread().getName() + "： onComplete ");
            }
        };
        //消费者订阅中间处理器，subscribe用来做关系绑定
        publisher.subscribe(subscriber);
        for (int i = 0; i < 10; i++) {
            //生产者生产内容
            publisher.submit("消息 : " + i);
        }
        //结束后关闭发布者
        //正式环境放finally里或用try-resource确保关闭
        publisher.close();


    }

    @RequestMapping("pubProSub")
    public void pubProSub() {
        //创建生产者
        SubmissionPublisher publisher = new SubmissionPublisher<>();
        // 创建中间处理器
        ReactiveProcessor reactiveProcessor = new ReactiveProcessor();
        //中间处理器订阅生产者,subscribe用来做关系绑定
        publisher.subscribe(reactiveProcessor);
        //创建消费者
        Flow.Subscriber subscriber = new Flow.Subscriber() {
            //定义成员变量来共享订阅关系
            Flow.Subscription subscription;

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                log.debug(Thread.currentThread().getName() + "：中间处理器与订阅者建立订阅关系，只会执行一次");
                //把订阅关系赋值给成员变量
                this.subscription = subscription;
                this.subscription.request(1);
            }

            @Override
            public void onNext(Object item) {
                this.subscription.request(1);
                log.debug(Thread.currentThread().getName() + "：receive :  " + item);
            }

            @Override
            public void onError(Throwable throwable) {
                //出现异常
                throwable.printStackTrace();
                //不再接收消息则调用cancel
                this.subscription.cancel();
                log.debug(Thread.currentThread().getName() + "： onError ");
            }

            @Override
            public void onComplete() {
                log.debug(Thread.currentThread().getName() + "： onComplete ");
            }


        };
        //消费者订阅中间处理器，subscribe用来做关系绑定
        reactiveProcessor.subscribe(subscriber);
        //生产者生产数据
        for (int i = 0; i < 10; i++) {
            publisher.submit("消息 : " + i);
        }
        //结束后关闭发布者
        //正式环境放finally里或用try-resource确保关闭
        publisher.close();

    }
    */
/*
     *
     * @author zgb
     * @desc 解决背压问题：中间处理器订阅发布者， 同时消费者再订阅中间处理器。中间处理器也可以调节发布订阅的生产消费速率。
     * @date 2022-11-24 16:57:46

     *//*


    class ReactiveProcessor extends SubmissionPublisher implements Flow.Subscriber {
        private Flow.Subscription subscription;

        @Override
        public void onSubscribe(Flow.Subscription subscription) {
            log.debug(Thread.currentThread().getName() + "：ReactiveProcessor 发布者与中间处理器建立订阅关系，只会执行一次");
            this.subscription = subscription;
            this.subscription.request(1);
        }

        @Override
        public void onNext(Object item) {
            log.debug(Thread.currentThread().getName() + "：ReactiveProcessor receive: " + item);
            this.submit(item.toString().toUpperCase());
            this.subscription.request(1);
        }

        @Override
        public void onError(Throwable throwable) {
            log.debug(" ReactiveProcessor error ");
            throwable.printStackTrace();
            this.subscription.cancel();
        }

        @Override
        public void onComplete() {
            log.debug(Thread.currentThread().getName() + "：ReactiveProcessor complete ");
        }
    }

}
*/
